package com.atguigu.flink.sqlfunction;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.Expressions;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * Created by Smexy on 2023/3/5
 *  UDTF: 输入1行1列，输出 N行N列。
 *          举例:  输入: a_1
 *                 输出:  a,3
 *                       1,3
 */
public class Demo3_TableFunction
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        String createTableSql = " create table t1 ( id string,ts bigint, vc int  )" +
            " with ( " +
            " 'connector' = 'filesystem' ,   " +
            " 'path' =  'data/cep1.txt' ,   " +
            "  'format' = 'csv' " +
            "      )                 ";

        //执行建表
        tableEnvironment.executeSql(createTableSql);
        Table table = tableEnvironment.from("t1");

        MyUDTF myUDTF = new MyUDTF();
        /*
            TableAPI中使用

            Hive中如何用UDTF?
                select xxx,col1,col2
                from xxx
                lateral view UDTF(xxx) tmp as col1,col2

                lateral view语法本质是 join。

             在 Table API 中，表值函数是通过 .joinLateral(...) 或者 .leftOuterJoinLateral(...) 来使用的

         */
        tableEnvironment.createTemporaryFunction("myUDTF",myUDTF);

        /*table
            //.joinLateral(Expressions.call("myUDTF",$("id")))
            .leftOuterJoinLateral(Expressions.call("myUDTF",$("id")))
             .select($("id"),$("word"),$("length"))
             .execute().print();*/



        //SQL中使用
        tableEnvironment.sqlQuery("select id,word,length from t1 " +
                              //  " LEFT JOIN LATERAL TABLE(myUDTF(id)) ON TRUE "
                                 "  JOIN LATERAL TABLE(myUDTF(id)) ON TRUE   "
        )
                        .execute()
                        .print();

    }

    /*
          TableFunction<T> ： T为 输出的一行的类型，如果有多列。要么封装为POJO，要么封装为Row(方便)

            不同于标量函数，表值函数的求值方法本身不包含返回类型，而是通过 collect(T) 方法来发送要输出的行

           @FunctionHint: 对函数的返回值类型，进行提示。提升编译的速度。
                           UDTF执行后返回的数据，就有了列名word,length
     */
    @FunctionHint(output = @DataTypeHint("ROW<word STRING, length INT>"))
    public static class MyUDTF extends TableFunction<Row>{

        //只能写void
        public void eval(String str){

            //如果是s3，输入后，调用函数，不输出
            if (!str.contains("s3")){
                String[] words = str.split("_");
                //输出一次，是输出一行
                collect(Row.of(words[0],str.length()));
                collect(Row.of(words[1],str.length()));
            }
        }
    }
}
